package com.wkh.datastructure.blockingqueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

//双锁实现阻塞队列
public class BlockingQueue2<E> implements BlockingQueue<E> {
    private E[] array;
    private int head;
    private int tail;
    //原子变量 --- 保证size的线程安全
    private AtomicInteger size = new AtomicInteger();

    public BlockingQueue2(int capacity) {
        array = (E[]) new Object[capacity];
    }

    //锁对象
    private ReentrantLock tailLock = new ReentrantLock();
    private ReentrantLock headLock = new ReentrantLock();
    //条件变量对象
    private Condition headWaits = headLock.newCondition();//与poll方法有关
    private Condition tailWaits = tailLock.newCondition();//与offer方法有关

    @Override
    public void offer(E value) throws InterruptedException{
        int c;//添加前元素个数
        //开锁
        try {
            tailLock.lockInterruptibly();
            while (isFull()) {//防止虚假唤醒
                tailWaits.await();
            }
            array[tail] = value;
            tail++;
            if (tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();//size++
            if (c+1<array.length){
                tailWaits.signal();
            }
            //唤醒等待非空的线程
        }finally {
            //解锁
            tailLock.unlock();
        }
        //防止死锁
        //唤醒等待非空的线程,唤醒要删除的线程
        if (c == 0){
            try {
                headLock.lockInterruptibly();
                headWaits.signal();
            }finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E value, long timeout) throws InterruptedException {
        int c;//添加前元素个数
        //开锁
        tailLock.lockInterruptibly();
        try {
            long t = TimeUnit.MILLISECONDS.toNanos(timeout);

            while (isFull()) {//防止虚假唤醒
                if (t<=0){
                    return false;
                }
                t = tailWaits.awaitNanos(t);
            }
            array[tail] = value;
            tail++;
            if (tail == array.length) {
                tail = 0;
            }
            //先获取原来的值再自增1
            c = size.getAndIncrement();//size++
        } finally {
            //解锁
            tailLock.unlock();
        }
        //防止死锁
        //唤醒等待非空的线程,唤醒要删除的线程
        if (c == 0){
            try {
                headLock.lockInterruptibly();
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        int c;//删除前元素个数
        E e;
        headLock.lockInterruptibly();
        try {
            //防止虚假唤醒
            while (isEmpty()) {
                headWaits.await();
            }
            e = array[head];
            array[head] = null;
            head++;
            if (head == array.length) {
                head = 0;
            }
            //获取之前的值再减1
            c = size.getAndDecrement();//size--
            //值还有富余，就唤醒等待的操作线程
            if (c>1){
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }
        //防止死锁
        //唤醒等待非满的线程
        if (c == array.length){
            try {
                tailLock.lockInterruptibly();
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    @Override
    public boolean isEmpty() {
        return size.get() == 0;
    }

    @Override
    public boolean isFull() {
        return size.get() == array.length;
    }

    @Override
    public String toString() {
        return "BlockingQueue1{" +
                "array=" + Arrays.toString(array) + '}';
    }
}
